package cn.le.listener;

import cn.le.enums.EDataFormat;
import cn.le.producer.KafkaMapping;
import cn.le.redis.BaseRedisUtil;
import cn.le.vo.*;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static cn.le.util.Constant.SOURCE_TOPIC;

@Component
public class SourceListener {


    @Autowired
    private KafkaMapping kafkaMapping;

    @Autowired
    private BaseRedisUtil redisUtil;

    @KafkaListener(topics = SOURCE_TOPIC)
    public void iotListener(String msg){
        System.out.println("-----------"+msg);
        KafkaSourceVO sourceVO = JSONObject.parseObject(msg,KafkaSourceVO.class);
        //设备信息
        RedisDeviceVO deviceVO = redisUtil.get(sourceVO.getImei());
        //产品信息
        RedisProductVO productVO = redisUtil.get(deviceVO.getProductId());
        if (EDataFormat.BYTE.getFormat().equals(productVO.getFormat())){
            analysisByte(sourceVO,productVO,deviceVO);
        }else if (EDataFormat.JSON.getFormat().equals(productVO.getFormat())){
            analysisJson(sourceVO,productVO,deviceVO);
        }
    }

    /**
     * 解析
     */
    public void analysisByte(KafkaSourceVO sourceVO,RedisProductVO productVO,RedisDeviceVO deviceVO){
        char[] chars = sourceVO.getData().toCharArray();
        List<RedisPropertyVO> propertys = productVO.getPropertys();

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = new ArrayList<>(propertys.size());

        propertys.forEach(property->{
            String[] str = property.getOfset().split("-");
            int begin = Integer.valueOf(str[0]);
            int end = Integer.valueOf(str[1]);

            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(property.getId());
            StringBuilder sb = new StringBuilder();
            for (int i = begin;i <= end; i++){
                sb.append(chars[i]);
            }
            data.setData(sb.toString());
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
        kafkaMapping.send(downVO);
    }

    /**
     * 解析json
     * @param sourceVO
     * @param productVO
     * @param deviceVO kafkaMapping.send(downVOList);
     */
    public void analysisJson(KafkaSourceVO sourceVO,RedisProductVO productVO,RedisDeviceVO deviceVO){
        Map<String,Long> propertyMap = productVO.getPropertys().stream().collect(Collectors.toMap(RedisPropertyVO :: getOfset,RedisPropertyVO::getId));
        Map<String,String> dataMap = JSONObject.parseObject(sourceVO.getData(), HashMap.class);

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = new ArrayList<>(dataMap.size());

        dataMap.forEach((key,val)->{
            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(propertyMap.get(key));
            data.setData(val);
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
        kafkaMapping.send(downVO);
    }
}
